package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.ExamLessonBean;
import com.atguigu.edu.realtime.common.bean.ExamPageBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.common.util.HBaseUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;

public class DwsExamLessonApp extends BaseApp {

    public static void main(String[] args) {
        new DwsExamLessonApp().start(13001,4,"DwsExamPageWindow", Constant.TOPIC_DWD_EXAMINATION_ANSWER_DETAIL);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonStream = kafkaStrDS.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> wmStream = jsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(
                        (v, ts) -> v.getLong("ts")*1000
                ));

        //jsonStream.print();

        SingleOutputStreamOperator<ExamLessonBean> processed = wmStream.keyBy(v -> v.getString("paper_id")).process(
                new ProcessFunction<JSONObject, ExamLessonBean>() {
                    @Override
                    public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, ExamLessonBean>.Context context, Collector<ExamLessonBean> collector) throws Exception {
                        ExamLessonBean examPageBean = ExamLessonBean.builder()
                                .paperId(jsonObject.getString("paper_id"))
                                .score(jsonObject.getDouble("score"))
                                .durationSec(jsonObject.getLong("duration_sec"))
                                .personNum(1L)
                                .ts(jsonObject.getLong("ts"))
                                .build();
                        collector.collect(examPageBean);
                    }
                }
        );




        //processed.print();
        SingleOutputStreamOperator<ExamLessonBean> resultStream = processed.keyBy(ExamLessonBean::getPaperId).window(
                TumblingEventTimeWindows.of(Time.seconds(10))
        ).reduce(
                new ReduceFunction<ExamLessonBean>() {
                    @Override
                    public ExamLessonBean reduce(ExamLessonBean examPageBean, ExamLessonBean t1) throws Exception {
                        long durationSec = examPageBean.getDurationSec() + t1.getDurationSec();
                        double score = examPageBean.getScore() + t1.getScore();
                        long personNum = examPageBean.getPersonNum() + t1.getPersonNum();
                        examPageBean.setDurationSec(durationSec);
                        examPageBean.setScore(score);
                        examPageBean.setPersonNum(personNum);
                        return examPageBean;
                    }
                },
                new WindowFunction<ExamLessonBean, ExamLessonBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<ExamLessonBean> iterable, Collector<ExamLessonBean> collector) throws Exception {
                        ExamLessonBean next = iterable.iterator().next();
                        next.setStt(DateFormatUtil.tsToDateTime(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.tsToDateTime(timeWindow.getEnd()));
                        next.setCurDate(DateFormatUtil.tsToDate(timeWindow.getStart()));
                        collector.collect(next);
                    }
                }
        );


        SingleOutputStreamOperator<ExamLessonBean> dimStream = resultStream.map(
                new RichMapFunction<ExamLessonBean, ExamLessonBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHbaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public ExamLessonBean map(ExamLessonBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度的主键
                        String paperId = orderBean.getPaperId();
                        //根据维度的主键获取对应的维度对象
                        JSONObject dimJsonObj = HBaseUtil.readRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_test_paper", paperId, JSONObject.class);
                        //将维度对象相关的维度属性补充到流中的对象上
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setCourseId(dimJsonObj.getString("course_id"));

                        return orderBean;
                    }
                }
        );

        SingleOutputStreamOperator<ExamLessonBean> dimCourseStream = dimStream.map(
                new RichMapFunction<ExamLessonBean, ExamLessonBean>() {
                    private Connection hbaseConn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        hbaseConn = HBaseUtil.getHbaseConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        HBaseUtil.closeHBaseConnection(hbaseConn);
                    }

                    @Override
                    public ExamLessonBean map(ExamLessonBean orderBean) throws Exception {
                        //根据流中的对象获取要关联的维度的主键
                        String courseId = orderBean.getCourseId();
                        //根据维度的主键获取对应的维度对象
                        JSONObject dimJsonObj = HBaseUtil.readRow(hbaseConn, Constant.HBASE_NAMESPACE, "dim_course_info", courseId, JSONObject.class);
                        //将维度对象相关的维度属性补充到流中的对象上
                        //id,spu_id,price,sku_name,sku_desc,weight,tm_id,category3_id,sku_default_img,is_sale,create_time
                        orderBean.setCourseName(dimJsonObj.getString("course_name"));

                        return orderBean;
                    }
                }
        );


        dimCourseStream.print();

        dimStream.map(new BeanToJsonStrMapFunction())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_exam_lesson_window"));
    }
}
